use std::io::{BufReader, BufWriter, Read, Write};

use log::{debug, error};

use crate::IOBUF_SIZE;

pub struct PipeReader(BufReader<PipeReaderInner>);

impl Read for PipeReader {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.0.read(buf)
    }
}

struct PipeReaderInner {
    // self.buf[self.top..] is a buffer for unread bytes.
    buf: Vec<u8>,
    top: usize,

    // Receive bytes from writer.
    buf_rx: std::sync::mpsc::Receiver<Vec<u8>>,
}

impl Read for PipeReaderInner {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }

        // Buffer in reader is not empty.
        if self.top < self.buf.len() {
            let n = buf.len().min(self.buf.len() - self.top);

            buf[..n].copy_from_slice(&self.buf[self.top..self.top + n]);

            self.top += n;

            return Ok(n);
        }

        // Buffer is empty.
        self.top = 0;
        self.buf.clear();

        match self.buf_rx.recv() {
            Ok(newbuf) => {
                self.buf.extend_from_slice(&newbuf);
                self.read(buf)
            }
            Err(_) => Ok(0),
        }
    }
}

pub struct PipeWriter(BufWriter<PipeWriterInner>);

impl Write for PipeWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.0.write(buf)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.0.flush()
    }
}

struct PipeWriterInner {
    buf_tx: std::sync::mpsc::Sender<Vec<u8>>,
}

impl Write for PipeWriterInner {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = buf.len();

        debug!("pipe: write {n} bytes");

        if n == 0 {
            return Ok(0);
        }

        match self.buf_tx.send(buf.to_vec()) {
            Ok(_) => Ok(n),
            Err(_) => {
                error!("pipe rx closed");
                Err(std::io::ErrorKind::WriteZero.into())
            }
        }
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

pub fn pipe() -> (PipeReader, PipeWriter) {
    let (tx, rx) = std::sync::mpsc::channel();

    let reader = PipeReaderInner {
        buf: vec![],
        top: 0,
        buf_rx: rx,
    };

    let writer = PipeWriterInner { buf_tx: tx };
    (
        PipeReader(BufReader::with_capacity(IOBUF_SIZE, reader)),
        PipeWriter(BufWriter::with_capacity(IOBUF_SIZE, writer)),
    )
}

#[test]
fn test_pipe() {
    let (mut reader, mut writer) = pipe();

    const N: usize = 10;
    const L: usize = 20;

    fn make_label(i: usize) -> Vec<u8> {
        let mut text = format!("<text {i:02}>").into_bytes();
        text.resize(L, 0);
        text
    }

    let reader_th = std::thread::spawn(move || {
        let mut buf = vec![0; L];

        for i in 0..N {
            reader.read(&mut buf).unwrap();

            assert_eq!(make_label(i), buf);
            println!("read {i:02}: {:?}", std::str::from_utf8(&buf));
        }
    });

    let writer_th = std::thread::spawn(move || {
        for i in 0..N {
            let text = make_label(i);

            writer.write(&text).unwrap();
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
    });

    let _ = reader_th.join();
    let _ = writer_th.join();
}
